package com.muh.operation.impl;

import co.elastic.clients.elasticsearch._types.SortOrder;
import co.elastic.clients.elasticsearch._types.aggregations.*;
import co.elastic.clients.elasticsearch._types.query_dsl.MultiMatchQuery;
import co.elastic.clients.elasticsearch._types.query_dsl.Query;
import co.elastic.clients.util.NamedValue;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.muh.entity.AggEmployee;
import com.muh.entity.Employee;
import com.muh.operation.AggEmployeeOperationService;
import com.muh.repository.EmployeeRepository;
import com.muh.util.Constent;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.domain.PageImpl;
import org.springframework.data.domain.PageRequest;
import org.springframework.data.domain.Pageable;
import org.springframework.data.domain.Sort;
import org.springframework.data.elasticsearch.client.elc.ElasticsearchAggregation;
import org.springframework.data.elasticsearch.client.elc.NativeQuery;
import org.springframework.data.elasticsearch.client.elc.NativeQueryBuilder;
import org.springframework.data.elasticsearch.core.ElasticsearchOperations;
import org.springframework.data.elasticsearch.core.SearchHit;
import org.springframework.data.elasticsearch.core.SearchHits;
import org.springframework.data.elasticsearch.core.query.HighlightQuery;
import org.springframework.data.elasticsearch.core.query.highlight.Highlight;
import org.springframework.data.elasticsearch.core.query.highlight.HighlightField;
import org.springframework.data.elasticsearch.core.query.highlight.HighlightFieldParameters;
import org.springframework.stereotype.Service;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;

@Service
public class AggEmployeeOperationServiceImpl implements AggEmployeeOperationService {
    @Autowired
    private EmployeeRepository employeeRepository;

    @Autowired
    private final ElasticsearchOperations elasticsearchOperations;

    public AggEmployeeOperationServiceImpl(ElasticsearchOperations elasticsearchOperations) {
        this.elasticsearchOperations = elasticsearchOperations;
    }

    @Override
    public PageImpl<AggEmployee> findPageHighlightByAddress(String address, int pageNumber, int pageSize) {
        //构建分页
        Pageable pageable = PageRequest.of(pageNumber, pageSize, Sort.Direction.DESC,"age");

        // 1. 构建多字段匹配查询
        Query query = MultiMatchQuery.of(m -> m
                .query(address)
                .fields("address") // title权重更高
        )._toQuery();

        // 2. 构建自定义高亮标签
        HighlightFieldParameters.HighlightFieldParametersBuilder builder = HighlightFieldParameters.builder();
        HighlightFieldParameters build = builder.withPreTags("<span style='color:red'>").withPostTags("</span>").build();
        HighlightField addressHighlight = new HighlightField("address", build);
        List<HighlightField> fields = new ArrayList<>();
        fields.add(addressHighlight);

        // 3. 配置高亮字段
        Highlight highlight = new Highlight(fields);

        // 4. 构建完整查询
        NativeQuery nativeQuery = new NativeQueryBuilder()
                .withQuery(query)
                .withHighlightQuery(new HighlightQuery(highlight, AggEmployee.class))
                .withPageable(pageable) // 设置分页
                .build();
        // 5. 执行查询
        SearchHits<AggEmployee> hits = elasticsearchOperations.search(nativeQuery, AggEmployee.class);
        ArrayList<AggEmployee> aggEmployees = new ArrayList<>();
//        List<AggEmployee> employees = hits.stream().map().collect(Collectors.toList());
//        List<SearchHit<Employee>> searchHits = hits.getSearchHits();
//        List<Employee> employees = searchHits.stream().map(SearchHit::getContent).collect(Collectors.toList());
        return new PageImpl<>(
                aggEmployees,
                pageable,
                hits.getTotalHits()
        );
    }

    private Employee processSearchHit(SearchHit<Employee> hit) {
        Employee employee = hit.getContent();

        // 处理标题高亮
        if (hit.getHighlightFields().containsKey("address")) {
            employee.setHighlightAddress(
                    hit.getHighlightFields().get("address").get(0)
            );
        }

//        // 处理内容高亮（多个字段）
//        if (hit.getHighlightFields().containsKey("content")) {
//            employee.setHighlightContent(
//                    hit.getHighlightFields().get("content")
//            );
//        }

        return employee;
    }


    @Override
    public Map<String, Object> metricAggregation(String address) {
        long b3 = System.currentTimeMillis();
        //构建多字段匹配查询 根据字段address查询，将字段address
        Query query = MultiMatchQuery.of(m -> m
                .query(address)
                .fields("address") // title权重更高
        )._toQuery();

//        NativeQuery withNativeQuery = new NativeQueryBuilder().withQuery(query).build();
//        NativeQuery withNativeQuery = new NativeQueryBuilder().build();
//        long count1 = elasticsearchOperations.count(withNativeQuery, Employee.class);//根据条件查询总记录数
        long e3 = System.currentTimeMillis();
        System.out.println("withNativeQuery耗时\t"+(e3-b3));

        long b4 = System.currentTimeMillis();
        // 1. 构建聚合 对字段 age 进行统计
        Aggregation statsAgg = Aggregation.of(a -> a
                .extendedStats(ExtendedStatsAggregation.of(e -> e.field("age"))));//聚合的方式 extendedStats
        //根据 sex 去重 统计
        Aggregation distinct = Aggregation.of(a -> a
                .cardinality(CardinalityAggregation.of(e -> e.field("sex"))));//聚合的方式 cardinality
        //根据 job 分组 统计数量
        Aggregation terms = Aggregation.of(a -> a
                .terms(TermsAggregation.of(e -> e.field("job.keyword"))));//聚合的方式 terms
        //根据 job 分组 统计数量 然后排序
        Aggregation termsJobThenOrderCount = Aggregation.of(a -> a
                .terms(
                        TermsAggregation.of(e -> e
                            .field("job.keyword")
                            .size(10)//显示10条数据
                            .order(
                                    new NamedValue<SortOrder>("_count", SortOrder.Asc)
                            )
                        )
                )
        );//聚合的方式 terms

        // 2. 构建查询
        NativeQuery withAggregationQuery = new NativeQueryBuilder()
//                .withQuery(query)//查询条件
                .withAggregation("alias_name", statsAgg)
                .withMaxResults(0)//不需要查询结果
                .build();
        //3、执行查询
        SearchHits<AggEmployee> searchHits = elasticsearchOperations.search(withAggregationQuery, AggEmployee.class);

        //4、封装数据
        Map<String, Object> stringObjectMap = Constent.extractAggregationResults(searchHits);

        long e4 = System.currentTimeMillis();
        System.out.println("withAggregationQuery耗时\t"+(e4-b4));
        return stringObjectMap;
    }

    @Override
    public JSONArray bucketAggregation(String address) {
        long b3 = System.currentTimeMillis();
        //构建多字段匹配查询 根据字段address查询，将字段address
        Query query = MultiMatchQuery.of(m -> m
                .query(address)
                .fields("address") // title权重更高
        )._toQuery();

//        NativeQuery withNativeQuery = new NativeQueryBuilder().withQuery(query).build();
//        NativeQuery withNativeQuery = new NativeQueryBuilder().build();
//        long count1 = elasticsearchOperations.count(withNativeQuery, Employee.class);//根据条件查询总记录数
        long e3 = System.currentTimeMillis();
        System.out.println("withNativeQuery耗时\t"+(e3-b3));

        long b4 = System.currentTimeMillis();
        // 1. 构建聚合 对字段 age 进行统计
        Aggregation statsAgg = Aggregation.of(a -> a
                .extendedStats(ExtendedStatsAggregation.of(e -> e.field("age"))));//聚合的方式 extendedStats
        //根据 sex 去重 统计
        Aggregation distinct = Aggregation.of(a -> a
                .cardinality(CardinalityAggregation.of(e -> e.field("sex"))));//聚合的方式 cardinality
        //根据 job 分组 统计数量
        Aggregation terms = Aggregation.of(a -> a
                .terms(TermsAggregation.of(e -> e.field("job.keyword"))));//聚合的方式 terms
        //根据 job 分组 统计数量 然后根据数量排序 排序只能选择_key和_count
        Aggregation termsJobThenOrderCount = Aggregation.of(a -> a
                .terms(
                        TermsAggregation.of(e -> e
                                .field("job.keyword")
                                .size(10)
                                .order(
                                        new NamedValue<SortOrder>("_count", SortOrder.Asc)
                                )
                        )
                )
        );//聚合的方式 terms

        // 2. 构建查询
        NativeQuery withAggregationQuery = new NativeQueryBuilder()
//                .withQuery(query)//查询条件
                .withAggregation("alias_name", termsJobThenOrderCount)//别名
                .withMaxResults(0)//不需要查询结果
                .build();
        //3、执行查询
        SearchHits<AggEmployee> searchHits = elasticsearchOperations.search(withAggregationQuery, AggEmployee.class);

        ArrayList aggregations = (ArrayList) searchHits.getAggregations().aggregations();
        ElasticsearchAggregation elasticsearchAggregation = (ElasticsearchAggregation) aggregations.get(0);
        StringTermsAggregate stringTermsAggregate = (StringTermsAggregate) elasticsearchAggregation.aggregation().getAggregate()._get();
        JSONArray jsonArray = new JSONArray();
        stringTermsAggregate.buckets().array().forEach(bucket -> {
            JSONObject jsonObject = new JSONObject();
            jsonObject.put("key", bucket.key().stringValue());
            jsonObject.put("count", bucket.docCount());
            jsonArray.add(jsonObject);
//            System.out.println(bucket.key().stringValue()+"<>"+bucket.docCount());
        });

        long e4 = System.currentTimeMillis();
        System.out.println("withAggregationQuery耗时\t"+(e4-b4));
        return jsonArray;
    }

}
